package com.kk.netty.basice.demo2;

import io.netty.channel.DefaultEventLoop;
import io.netty.channel.EventLoop;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;


@Slf4j
public class FutureAndPromise {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        test11 ( );
    }

    public static void test6() throws ExecutionException, InterruptedException {
        DefaultEventLoop eventExecutors = new DefaultEventLoop ( );
        DefaultPromise<Integer> promise = new DefaultPromise<> (eventExecutors);

        // submit 提交任务，返回值是 Future
        eventExecutors.submit (() -> {
            System.out.println ("1");
            try {
                promise.await ( );
                // 注意不能仅捕获 InterruptedException 异常
                // 否则 死锁检查抛出的 BlockingOperationException 会继续向上传播
                // 而提交的任务会被包装为 PromiseTask，它的 run 方法中会 catch 所有异常然后设置为 Promise 的失败结果而不会抛出
            } catch (Exception e) {
                e.printStackTrace ( );
            }
            System.out.println ("2");
        });
        eventExecutors.submit (() -> {
            System.out.println ("3");
            try {
                promise.await ( );
            } catch (Exception e) {
                e.printStackTrace ( );
            }
            System.out.println ("4");
        });
    }

    public static void test5() throws ExecutionException, InterruptedException {
        DefaultEventLoop eventExecutors = new DefaultEventLoop ( );
        DefaultPromise<Integer> promise = new DefaultPromise<> (eventExecutors);

        promise.addListener (future -> {
            log.debug ("result {}", (promise.isSuccess ( ) ? promise.getNow ( ) : promise.cause ( )).toString ( ));
        });

        eventExecutors.execute (() -> {
            try {
                Thread.sleep (1000);
            } catch (InterruptedException e) {
                e.printStackTrace ( );
            }
            RuntimeException e = new RuntimeException ("error...");
            log.debug ("set failure, {}", e.toString ( ));
            promise.setFailure (e);
        });

        log.debug ("start...");
    }

    public static void test4() throws ExecutionException, InterruptedException {
        DefaultEventLoop eventExecutors = new DefaultEventLoop ( );
        DefaultPromise<Integer> promise = new DefaultPromise<> (eventExecutors);

        eventExecutors.execute (() -> {
            try {
                Thread.sleep (1000);
            } catch (InterruptedException e) {
                e.printStackTrace ( );
            }
            RuntimeException e = new RuntimeException ("error...");
            log.debug ("set failure, {}", e.toString ( ));
            promise.setFailure (e);
        });

        log.debug ("start...");
        log.debug ("getNow:{}", promise.getNow ( ));
        promise.await ( ); // 与 sync 和 get 区别在于，不会抛异常
        log.debug ("result {}", (promise.isSuccess ( ) ? promise.getNow ( ) : promise.cause ( )).toString ( ));
    }

    public static void test3() throws ExecutionException, InterruptedException {
        DefaultEventLoop eventExecutors = new DefaultEventLoop ( );
        DefaultPromise<Integer> promise = new DefaultPromise<> (eventExecutors);

        eventExecutors.execute (() -> {
            try {
                Thread.sleep (1000);
            } catch (InterruptedException e) {
                e.printStackTrace ( );
            }
            RuntimeException e = new RuntimeException ("error...");
            log.debug ("set failure, {}", e.toString ( ));
            promise.setFailure (e);
        });

        log.debug ("start...");
        log.debug ("getNow:{}", promise.getNow ( ));
        promise.get ( ); // sync() 也会出现异常，只是 get 会再用 ExecutionException 包一层异常
    }

    public static void test2() {
        DefaultEventLoop eventExecutors = new DefaultEventLoop ( );
        DefaultPromise<Integer> promise = new DefaultPromise<> (eventExecutors);

        // 设置回调，异步接收结果
        promise.addListener (future -> {
            // 这里的 future 就是上面的 promise
            log.debug ("getNow:{}", future.getNow ( ));
        });

        eventExecutors.execute (() -> {
            try {
                Thread.sleep (1000);
            } catch (InterruptedException e) {
                e.printStackTrace ( );
            }
            log.debug ("set success, {}", 10);
            promise.setSuccess (10);
        });
        log.debug ("start...");
    }

    public static void test1() throws ExecutionException, InterruptedException {
        DefaultEventLoop eventExecutors = new DefaultEventLoop ( );
        DefaultPromise<Integer> promise = new DefaultPromise<> (eventExecutors);

        eventExecutors.execute (() -> {
            try {
                Thread.sleep (1000);
            } catch (InterruptedException e) {
                e.printStackTrace ( );
            }
            log.debug ("set success, {}", 10);
            promise.setSuccess (10);
        });

        log.debug ("start...");
        log.debug ("getNow:{}", promise.getNow ( )); // 还没有结果
        log.debug ("get:{}", promise.get ( ));
    }

    // netty-future()
    public static void test11() throws ExecutionException, InterruptedException {
        NioEventLoopGroup group = new NioEventLoopGroup ( );
        EventLoop eventLoop = group.next ( );
        Future<Integer> future = eventLoop.submit (new Callable<Integer> ( ) {
            @Override
            public Integer call() throws InterruptedException {
                log.info ("执行计算！");
                Thread.sleep (1000);
                return 50;
            }
        });
        log.info ("等待结果中......");

        //>>>>>>>>>>>>同步>>>>>>>>>>>>>>>>>
        // 控制台打印是main线程打印
//        log.info ("结果为：{}", future.get ( ));
        //>>>>>>>>>>>>同步>>>>>>>>>>>>>>>>>

        //>>>>>>>>>>>>异步>>>>>>>>>>>>>>>>>
        future.addListener (new GenericFutureListener<Future<? super Integer>> ( ) {
            @Override
            public void operationComplete(Future<? super Integer> future) throws Exception {
                // 控制台打印是 nioEventLoopGroup 线程组打印
                log.info ("结果为：{}", future.getNow ( ));
            }
        });
        //>>>>>>>>>>>>异步>>>>>>>>>>>>>>>>>

    }

    // jdk-future
//    public static void test() throws ExecutionException, InterruptedException {
//        // 1.线程池
//        ExecutorService executorService = Executors.newFixedThreadPool (2);
//        // 2.提交任务
//        Future<Integer> future = executorService.submit (new Callable<Integer> ( ) {
//            @Override
//            public Integer call() throws InterruptedException {
//                log.info ("执行计算！");
//                Thread.sleep (1000);
//                return 50;
//            }
//        });
//        // 3、主线程通过 future获取结果
//        log.info ("等待结果中......");
//        log.info ("结果为：{}", future.get ( ));
//    }
}
